package com.xzc.apitest.processFunction

import com.xzc.apitest.source.SensorReading
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.util.Collector

object ProcessFunctionTest {
  def main(args: Array[String]): Unit = {
    //创建执行环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val inputStream = env.socketTextStream("hadoop102", 7777);

    val dataStream = inputStream
      .map(data => {
        val arr = data.split(",")
        SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
      })
    //      .keyBy(_.id)
    //      .process(new MyKeyedProcessFunction)

    //需求：温度10秒内连续上升
    //用窗口函数来做,不好搞，因为窗口是有界的，有界会打断连续
    //    val warningStream = dataStream
    //      .keyBy(_.id)
    //      .timeWindow(Time.seconds(10))
    //用定时器，在数据标记后开始计时
    val warningStream = dataStream
      .keyBy(_.id)
      .process(new TempIncreaseWarning(10000L))

    warningStream.print()

    env.execute("process function test")
  }
}

class TempIncreaseWarning(interval: Long) extends KeyedProcessFunction[String, SensorReading, String] {
  //定义状态：保存上一个温度值，进行比较
  lazy val lastTempState: ValueState[Double] = getRuntimeContext.getState(
    new ValueStateDescriptor[Double]("last-temp", classOf[Double])
  )
  //注册定时器时间戳用于删除
  lazy val timerTsState: ValueState[Long] = getRuntimeContext.getState(
    new ValueStateDescriptor[Long]("timer-ts", classOf[Long])
  )

  override def processElement(value: SensorReading, ctx: KeyedProcessFunction[String, SensorReading, String]#Context, out: Collector[String]): Unit = {
    //先取状态
    val lastTemp = lastTempState.value()
    val timerTs = timerTsState.value()

    //更新温度值
    lastTempState.update(value.temperature)

    //当前温度值和上次温度进行比较
    if (value.temperature > lastTemp && timerTs == 0) {
      //如果温度上升，且还没有定时器，那么注册当前时间10秒后-定时器
      val ts = ctx.timerService().currentProcessingTime() + interval
      //这里取的是processing time就是按系统的时间来，而非数据时间
      ctx.timerService().registerProcessingTimeTimer(ts)
      timerTsState.update(ts)
    } else if (value.temperature < lastTemp) {
      //如果温度下降，那么删除定时器
      ctx.timerService().deleteProcessingTimeTimer(timerTs)
      timerTsState.clear()
    }
  }

  //触发定时器
  override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[String, SensorReading, String]#OnTimerContext, out: Collector[String]): Unit = {
    //只要没有关闭说明温度一直在上升
    out.collect("传感器" + ctx.getCurrentKey + "的温度连续" + interval / 1000 + "秒持续上升")
    timerTsState.clear()
  }
}

class MyKeyedProcessFunction extends KeyedProcessFunction[String, SensorReading, String] {
  var myState: ValueState[Int] = _


  override def open(parameters: Configuration): Unit = {
    myState = getRuntimeContext.getState(new ValueStateDescriptor[Int]("mystate", classOf[Int]))
  }

  override def processElement(value: SensorReading,
                              ctx: KeyedProcessFunction[String, SensorReading, String]#Context,
                              out: Collector[String]): Unit = {
    ctx.getCurrentKey
    ctx.timestamp()
    ctx.timerService().currentWatermark()
    //以时间戳来定义
    val time1: Long = ctx.timestamp() + 60 * 1000L
    ctx.timerService().registerEventTimeTimer(time1)
    ctx.timerService().deleteEventTimeTimer(time1)
  }

  override def onTimer(
                        timestamp: Long,
                        ctx: KeyedProcessFunction[String, SensorReading, String]#OnTimerContext,
                        out: Collector[String]): Unit = {

  }
}
